package cqupt.spark_log.runable;

import java.util.ArrayList;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CountDownLatch;

import cqupt.spark_log.contrant.Contant;
import cqupt.spark_log.entity.Job;
import cqupt.spark_log.entity.Stage;
import cqupt.spark_log.entity.Task;
import cqupt.spark_log.fileio.FileReadMan;
import cqupt.spark_log.info.BanchInfo;
import cqupt.spark_log.utils.ECUtil;
import cqupt.spark_log.utils.Utils;


public class Monitor implements Runnable {
	private BanchInfo banchInfo;
	private FileReadMan fileTools;
	private String executorId;
	private CountDownLatch countDownLatch;
	private double baseEnergy;
	public Monitor(BanchInfo banchInfo, String fileName, CountDownLatch countDownLatch) {
		this.banchInfo = banchInfo;
		this.fileTools = new FileReadMan("/monitor/" + fileName,true);
		this.executorId = Contant.getExecutorId(fileName);
		this.countDownLatch = countDownLatch;
		this.baseEnergy = Contant.getBaseEnergy(fileName);
	}

	public void run() {
		List<String> contentList = new ArrayList<String>();
		while (fileTools.hasNext()) {
			String content = fileTools.next();
			contentList.add(content);
		}
		Map<String, Job> jobs = banchInfo.getJobs();
		for (String key : jobs.keySet()) {
			Job job = jobs.get(key);
			for (Stage stage : job.getStages()) {
				Date stageStart = stage.getStartTime();
				Date stageEnd = stage.getEndTime();
				Date lastPoint = null;
				Date currentPoint = null;
				Map<String,Double> ecs = new HashMap<String, Double>();
				for (int i=0; i<contentList.size(); i++) {
					String content = contentList.get(i);
					if (content.matches("time.*")) {
						Date point = Utils.getMonitorTime(content);
						if (lastPoint == null) {
							lastPoint = new Date(point.getTime()-1000);
						} else {
							lastPoint = currentPoint;
						} 
						currentPoint = point;
					} else {
						if (currentPoint != null && (currentPoint.after(stageStart) || currentPoint.equals(stageStart))
								&& (currentPoint.before(stageEnd) || currentPoint.equals(stageEnd))) {
							String[] contents = content.split(";");
							String pid = contents[0];
							double cpu = Double.parseDouble(contents[1]);
							double memorey = Double.parseDouble(contents[2]);
							double ecPoint = ECUtil.getEc(lastPoint,currentPoint,cpu,memorey);
							if (ecs.containsKey(pid)){
								double ec = ecs.get(pid);
								double sum = ec+ecPoint;
								ecs.put(pid,ECUtil.get4point(sum));
							} else {
								ecs.put(pid,ecPoint);
							}
						} else {
							continue;
						}
					}
				}
				double stageEc = baseEnergy;
				for (String pid: ecs.keySet()) {
					stageEc += ecs.get(pid);
				}
				if (!ecs.isEmpty()) {
					stageEc = (stageEc/ecs.size());
				}
				stageEc = ECUtil.get4point(stageEc);
//				System.out.println("进程["+executorId+"]分析stage的能耗[" + stage.getStageId() + "]为["+stageEc+"]");
				double sumTime = 0;
				for (Task task: stage.getTasks()) {
					String executor = task.getExecutorId();
					if (this.executorId.equals(executor)) {
						sumTime += task.getRunTime();
					}
				}//end task for
				for (Task task: stage.getTasks()) {
					String executor = task.getExecutorId();
					if (this.executorId.equals(executor)) {
						double taskEc = stageEc*(task.getRunTime()/sumTime);
						task.setEc(ECUtil.get4point(taskEc));
					}
				}//end task for
			}//end stage for
		}//end job for
		countDownLatch.countDown();
	}
}
